package com.zhangpan.realtime.dim.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.zhangpan.realtime.common.base.BaseApp;
import com.zhangpan.realtime.common.bean.TableProcessDim;
import com.zhangpan.realtime.common.constant.Constant;
import com.zhangpan.realtime.common.util.HBaseUtil;
import com.zhangpan.realtime.common.util.JdbcUtil;
import com.zhangpan.realtime.dim.function.HBaseSinkFunction;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.io.IOException;
import java.util.*;

@Slf4j
public class DimApp extends BaseApp {

    public static void main(String[] args) {

        new DimApp().start(10001,4,"dim_app", Constant.TOPIC_DB);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        log.info("1. 对消费的数据做数据清洗");
        SingleOutputStreamOperator<JSONObject> etlStream = etl(stream);

        //2. 通过flink cdc读取配置表的信息
        log.info("2. 通过flink cdc读取配置表的信息");
        SingleOutputStreamOperator<TableProcessDim> tpStream = readTableProcess(env);

        //3. 根据配置表的数据 再hbase中建表
        log.info("3. 根据配置表的数据 再hbase中建表");
        tpStream = createHBaseTable(tpStream);

        //4. 数据流去connect 配置流
        log.info("4. 数据流去connect 配置流");
        SingleOutputStreamOperator<Tuple2<JSONObject,TableProcessDim>> dimDataToTpStream = connect(etlStream,tpStream);

        //5 删除不需要的字段
        log.info("5 删除不需要的字段");
        SingleOutputStreamOperator<Tuple2<JSONObject,TableProcessDim>> resultStream = deleteNotNeedColumns(dimDataToTpStream);

        //6. 写出到hbase 中
        log.info("6. 写出到hbase 中");
        writeToBase(resultStream);

    }

    private void writeToBase(SingleOutputStreamOperator<Tuple2<JSONObject,TableProcessDim>> resultStream) {

        /**
         *
         * 1. 有没有专门的Hbase 连接器
         * 2. sql 有专门Hbase连接器 由于一次只能写到一个表中  所以也不能把流转换位表在写
         * 3. 自定义sink
         */

        resultStream.addSink(new HBaseSinkFunction());
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                try {
                    JSONObject obj = JSON.parseObject(s);
                    String db = obj.getString("database");
                    String type = obj.getString("type");
                    String data = obj.getString("data");
                    return "gmall2025".equals(db) && ("insert".equals(type) || "update".equals(type) || "delete".equals(type)
                            ||"bootstrap-insert".equals(type)) && data != null && data.length() >2;

                }catch (Exception e) {
                    System.out.println("这不是一个正确的格式的数据");
                    log.warn("不是一个正确的json格式数据:{}",s);
                    return false;
                }
            }
        }).map(JSON::parseObject);

    }


    private SingleOutputStreamOperator<TableProcessDim> readTableProcess(StreamExecutionEnvironment env) {

        // useSSL = false;
        Properties props = new Properties();
        props.setProperty("useSSL","false");
        props.setProperty("allowPublicKeyRetrieval","true");
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .port(Constant.MYSQL_PORT)
                .databaseList("gmall2025_config")
                .tableList("gmall2025_config.table_process_dim")
                .username(Constant.MYSQL_USER_NAME)
                .password(Constant.MYSQL_PASSWORD)
                .jdbcProperties(props)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();
        return env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"cdc-source")
                .setParallelism(1)
                .map(new MapFunction<String, TableProcessDim>() {
                    @Override
                    public TableProcessDim map(String s) throws Exception {
                        JSONObject obj = JSON.parseObject(s);
                        String op = obj.getString("op");
                        TableProcessDim tableProcessDim;
                        if ("d".equals(op)) {
                            tableProcessDim = obj.getObject("before", TableProcessDim.class);
                        } else {
                            tableProcessDim = obj.getObject("after", TableProcessDim.class);
                        }
                        tableProcessDim.setOp(op);
                        return tableProcessDim;
                    }
                }).setParallelism(1);

    }

    private SingleOutputStreamOperator<TableProcessDim> createHBaseTable(SingleOutputStreamOperator<TableProcessDim> tpStream) {
        return tpStream.map(new RichMapFunction<TableProcessDim, TableProcessDim>() {

            private Connection hbaseConn;

            @Override
            public void open(Configuration parameters) throws Exception {
                //1. 获取Hbase 链接
                hbaseConn = HBaseUtil.getHBaseConnection();
            }

            @Override
            public void close() throws Exception {
                //2. 关闭链接
                HBaseUtil.closeHBaseConn(hbaseConn);
            }

            @Override
            public TableProcessDim map(TableProcessDim tableProcessDim) throws Exception {
                String op = tableProcessDim.getOp();
                if ("d".equals(op)) {
                    dropTable(tableProcessDim);
                } else if ("r".equals(op) || "c".equals(op)) {
                    createTable(tableProcessDim);
                } else {
                    dropTable(tableProcessDim);
                    createTable(tableProcessDim);
                }
                return tableProcessDim;

            }

            private void createTable(TableProcessDim tableProcessDim) throws IOException {
                // namespace
                HBaseUtil.createHBaseTable(hbaseConn,Constant.HBASE_NAMESPACE, tableProcessDim.getSinkTable(), tableProcessDim.getSinkFamily());
            }

            private void dropTable(TableProcessDim tableProcessDim) throws IOException {
                HBaseUtil.dropHBaseTable(hbaseConn,Constant.HBASE_NAMESPACE,tableProcessDim.getSinkTable());
            }

        }).setParallelism(1);
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject,TableProcessDim>> connect(SingleOutputStreamOperator<JSONObject> dataStream,
                                                                                   SingleOutputStreamOperator<TableProcessDim> configStream) {
        MapStateDescriptor<String,TableProcessDim> mapStateDescriptor = new MapStateDescriptor<>("table_process_dim",String.class, TableProcessDim.class);
        BroadcastStream<TableProcessDim> broadcastStream = configStream.broadcast(mapStateDescriptor);
        //2. 数据流去connect 广播流
        return dataStream.connect(broadcastStream)
                .process(new BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>() {

                    private HashMap<String,TableProcessDim> map;


                    @Override
                    public void open(Configuration parameters) throws Exception {

                        // open 中没办法访问状态
                        map = new HashMap<>();
                        //1. 去mysql 中查询table_process 表所有的数据
                        java.sql.Connection mysqlConn = JdbcUtil.getMysqlConnection();
                        List<TableProcessDim> tableProcessDimList = JdbcUtil.queryList(mysqlConn,"select * from gmall2025_config.table_process_dim",
                                TableProcessDim.class,true);
                        for (TableProcessDim tableProcessDim:tableProcessDimList) {
                            String key = tableProcessDim.getSourceTable();
                            map.put(key,tableProcessDim);
                        }
                        JdbcUtil.closeConnection(mysqlConn);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, ReadOnlyContext context, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {

                        ReadOnlyBroadcastState<String, TableProcessDim> state = context.getBroadcastState(mapStateDescriptor);
                        String key = jsonObject.getString("table");
                        TableProcessDim tableProcessDim = state.get(key);

                        if (tableProcessDim == null) {
                            tableProcessDim = map.get(key);
                            if (tableProcessDim != null) {
                                log.info("在map 中查找到:" + key);
                            }
                        } else {
                            log.info("在状态中找到" + key) ;
                        }

                        if (tableProcessDim != null ) {
                            log.info("当前数据#######:{}", jsonObject.toJSONString());
                            JSONObject data = jsonObject.getJSONObject("data");
                            data.put("op_type",jsonObject.getString("type"));
                            out.collect(Tuple2.of(data,tableProcessDim));
                        }


                    }

                    @Override
                    public void processBroadcastElement(TableProcessDim tableProcessDim, Context context, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {

                        BroadcastState<String, TableProcessDim> state = context.getBroadcastState(mapStateDescriptor);
                        String key = tableProcessDim.getSourceTable();
                        if ("d".equals(tableProcessDim.getOp())) {
                            // 删除状态
                            state.remove(key);
                            map.remove(key);
                        } else {
                            state.put(key,tableProcessDim);
                        }

                    }
                });


    }

    private SingleOutputStreamOperator<Tuple2<JSONObject,TableProcessDim>> deleteNotNeedColumns (SingleOutputStreamOperator<Tuple2<JSONObject,TableProcessDim>> dimDataToTpStream) {
        return dimDataToTpStream.map(new MapFunction<Tuple2<JSONObject, TableProcessDim>, Tuple2<JSONObject, TableProcessDim>>() {
            @Override
            public Tuple2<JSONObject, TableProcessDim> map(Tuple2<JSONObject, TableProcessDim> dataWithConfig) throws Exception {
                JSONObject data = dataWithConfig.f0;
                List<String> columns = new ArrayList<>(Arrays.asList(dataWithConfig.f1.getSinkColumns().split(",")));

                log.info("要处理的数据:{}",data.toJSONString());
                log.info("字段列表:{}", ArrayUtils.toString(columns));
                columns.add("op_type");
                data.keySet().removeIf(key -> !columns.contains(key));
                return dataWithConfig;
            }
        });

    }
}
